package com.example.springbootelasticsearch.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;

/**
 * @Description: canal客户端链接：mysql --> canal-server --> canal-client --> es
 * @Author zhangdongkang
 * @Date 2022/9/27
 */
@Slf4j
@Configuration
@ConditionalOnExpression("${canal.enabled}")
public class CanalClient implements DisposableBean {

	@Value("${canal.server.host}")
	private String serverHost;
	@Value("${canal.server.destination}")
	private String serverDestination;
	@Value("${canal.server.port}")
	private Integer serverPort;
	@Value("${canal.server.username}")
	private String serverUsername;
	@Value("${canal.server.password}")
	private String serverPassword;
	@Value("${canal.client.batchSize}")
	private Integer clientBatchSize;
	@Value("${canal.client.retryTimes}")
	private Integer clientRetryTimes;

	private CanalConnector canalConnector;

	/**
	 * canal客户端链接配置
	 */
	@Bean
	public CanalConnector getCanalConnector() {
		canalConnector = CanalConnectors.newClusterConnector(Lists.newArrayList(
										new InetSocketAddress(serverHost, serverPort)),
						serverDestination,
						serverUsername,
						serverPassword
		);
		canalConnector.connect();
		// 指定filter，格式{database}.{table}，不传参数就是 subscribe 所有的内容
		canalConnector.subscribe();
		// 回滚寻找上次中断的位置
		canalConnector.rollback();
		log.info("############ canal connect success , server host:{}", serverHost);
		return canalConnector;
	}

	/**
	 * 容器销毁，释放canal客户端链接
	 */
	@Override
	public void destroy() throws Exception {
		if (canalConnector != null) {
			canalConnector.disconnect();
			log.info("############ canal connect destroyed ...");
		}
	}

}
